package com.blueskykong.tm.common.netty.serizlize.dubbo;

import com.blueskykong.tm.common.netty.MessageCodecService;
import com.blueskykong.tm.common.netty.serizlize.hessian.HessianCodecServiceImpl;
import com.blueskykong.tm.common.netty.serizlize.kryo.KryoCodecServiceImpl;
import com.blueskykong.tm.common.netty.serizlize.kryo.KryoPoolFactory;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;
import java.util.Objects;

/**
 * @Author: Jason Chen
 */
public class DubboProtocolDecoder extends ByteToMessageDecoder {

    private static final int DUBBO_HEAD_LENGTH = 16;
    private static final short DUBBO_MAGIC = (short) 0xdabb;
    private static final byte SERIALIZATION_MASK = 0xf;

    private MessageCodecService util;

    public DubboProtocolDecoder() {
    }

    @Override
    protected void decode(ChannelHandlerContext cxt, ByteBuf in, List<Object> outList) throws Exception {
        int readableBytes = in.readableBytes();
        // 长度没有16字节继续等待
        if (readableBytes < DUBBO_HEAD_LENGTH) {
            return;
        }

        // 非dubbo协议头，关闭链路
        if (!checkDubboHead(in)) {
            cxt.close();
        }

        byte sId = (byte) (in.readByte() & SERIALIZATION_MASK);// 获取dubbo头序列化方式
        if (!setSerialization(sId)) {
            cxt.close();
        }
        int dataLength = in.readerIndex(in.readerIndex() + 9).readInt();
        if (dataLength == 0) {//没有数据，继续等待
            return;
        }

        int realLength = in.readableBytes();
        if (realLength < dataLength) {//数据不够，继续等待
            return;
        }

        byte[] dataBytes = new byte[dataLength];
        in.readBytes(dataBytes);
        outList.add(util.decode(dataBytes));
    }

    private boolean checkDubboHead(ByteBuf byteBuf) {
        short dubboMagic = byteBuf.readShort();
        return dubboMagic == DUBBO_MAGIC;
    }

    private boolean setSerialization(byte sId) {
        switch (sId) {
            case 2:
                util = new HessianCodecServiceImpl();
                break;
            case 8:
                util = new KryoCodecServiceImpl(KryoPoolFactory.getKryoPoolInstance());
                break;
            default:
                break;
        }
        return Objects.nonNull(util);
    }
}
